// @Author EthanScriptOn
// @Desc
package operator

import (
	"gitee.com/fatzeng/srf_switch_basic_components/util"
	"sync"
)

var dataPipelineMu = sync.Mutex{}
var dataPipelineContainer = make(map[interface{}]chan interface{})

func GetPipeline(operatorIdentifier interface{}) (resp chan interface{}) {
	_ = util.LockWithSync(dataPipelineMu.Lock, dataPipelineMu.Unlock, func() (err error) {
		resp = dataPipelineContainer[operatorIdentifier]
		return
	})
	return
}

func SetPipeline(operatorIdentifier interface{}, pipeline chan interface{}) {
	_ = util.LockWithSync(dataPipelineMu.Lock, dataPipelineMu.Unlock, func() (err error) {
		dataPipelineContainer[operatorIdentifier] = pipeline
		return
	})
}

func DestroyPipeline(operatorIdentifier interface{}) {
	_ = util.LockWithSync(dataPipelineMu.Lock, dataPipelineMu.Unlock, func() (err error) {
		pipeline := GetPipeline(operatorIdentifier)
		if pipeline == nil {
			return
		}
		close(pipeline)
		delete(dataPipelineContainer, operatorIdentifier)
		return
	})
}
